package com.concurrent.frame.processor.task;

import com.concurrent.frame.core.ResourceRegisterPool;
import com.concurrent.frame.pojo.JobInfo;
import com.concurrent.frame.pojo.TaskResult;
import com.concurrent.frame.pojo.TaskResultType;
import com.concurrent.frame.util.Assert;

import java.util.concurrent.Callable;


public abstract class AbsTaskProConProcessor<PT,PR,CR> implements ITaskProcessor<PT,PR> {
    protected JobInfo<CR> jobInfo;
    public AbsTaskProConProcessor(JobInfo<CR> jobInfo) {
        this.jobInfo = jobInfo;
    }
    public void process(final PT t){
        Callable<TaskResult> callable = new Callable<TaskResult>() {
            @Override
            public TaskResult<PR> call() throws Exception {
                TaskResult<PR> result;
                try {
                    result = taskProductExecute(t);
                    result = Assert.isTaskResult(result);
                } catch (Exception e) {
                    e.printStackTrace();
                    result = new TaskResult<PR>(TaskResultType.Exception, null, e.getMessage());
                }
                return result;
            }
        };
        jobInfo.getCompleteService().submit(callable);
        ResourceRegisterPool.getTaskExecutor().submit(new Runnable() {
            @Override
            public void run() {
                TaskResult<CR> result = null;
                try {
                    result = taskConsumerExecute( jobInfo.getCompleteService().take().get());
                    result = Assert.isTaskResult(result);
                }catch (Exception e){
                    e.printStackTrace();
                    result = new TaskResult<CR>(TaskResultType.Exception,null,e.getMessage());
                }finally {
                    jobInfo.addTaskResult(result);
                }
            }
         });

    }
    public abstract TaskResult<CR> taskConsumerExecute(TaskResult<PR> t);
    public abstract TaskResult<PR> taskProductExecute(PT t);

}
